package com.ylkj.bus.app.report;

import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.util.List;

/**
 * @Author: wengzx
 * @Date: 2025/3/29
 * @Desc: 处理t_iotmo_app表数据(先删除后插入)
 */
@Slf4j
public class UnpackIotpdeviceConDeleteAndInsert extends ProcessFunction<List<DeviceContainerDO>,List<DeviceContainerDO>> {

    //基础删除SQL模板
    private static final String BASE_DELETE_SQL = "DELETE FROM {0} WHERE device_code = ? ";

    private static final String INSERT_SQL =
            "INSERT INTO {0} (`id`, `container_name`, `image_version`, `create_status`, `container_status`, `device_code`, `is_remove`, `cpu_rate`, `memory_rate`,`storage_rate`,`source`,`ip`,`started`,`life_time`,`image`,`create_time`,`update_time`) " +
                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`container_name` = VALUES(`container_name`), " +
                    "`image_version` = VALUES(`image_version`), " +
                    "`create_status` = VALUES(`create_status`), " +
                    "`container_status` = VALUES(`container_status`), " +
                    "`device_code` = VALUES(`device_code`), " +
                    "`is_remove` = VALUES(`is_remove`), " +
                    "`cpu_rate` = VALUES(`cpu_rate`), " +
                    "`memory_rate` = VALUES(`memory_rate`), " +
                    "`storage_rate` = VALUES(`storage_rate`), " +
                    "`source` = VALUES(`source`), " +
                    "`ip` = VALUES(`ip`), " +
                    "`started` = VALUES(`started`), " +
                    "`life_time` = VALUES(`life_time`), " +
                    "`image` = VALUES(`image`), " +
                    "update_time = now()";

    // 数据库配置和表名
    private final RdsConfig rdsConfig;
    private final String tableName;

    private RdsUtils rdsUtils;

    // 格式化后的基础SQL语句
    private String baseSql;

    private String insertOnDuplicateSql;

    private static final int DEADLOCK_ERROR_CODE = 1213;
    //构造函数，用于初始化
    public UnpackIotpdeviceConDeleteAndInsert(String tableName , RdsConfig rdsConfig) {
        this.rdsConfig = rdsConfig;
        this.tableName = tableName;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);
        baseSql = MessageFormat.format(BASE_DELETE_SQL, tableName);
        insertOnDuplicateSql = MessageFormat.format(INSERT_SQL, tableName);
    }

    @Override
    public void processElement(List<DeviceContainerDO> deviceContainerDOs, Context ctx, Collector<List<DeviceContainerDO>> out) throws Exception {
        if (deviceContainerDOs == null || deviceContainerDOs.isEmpty()) {
            return;
        }
        DeviceContainerDO containerDO = deviceContainerDOs.get(0);
        String deviceCode = containerDO.getDeviceCode();
        int retryCount = 0;
        try {
            retryCount = 0;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        while (true) {
            try (Connection connection = rdsUtils.getConnection()) {
                connection.setAutoCommit(false); // 开启事务
                try {
                    boolean existsAndDeleted = deleteIfExists(connection, deviceCode);
                    if (existsAndDeleted) {
                        log.info("已删除旧记录，deviceCode: {}", deviceCode);
                    }
                    int insertedRows = insertBatchData(connection, deviceContainerDOs);
                    log.info("成功插入新记录，deviceCode: {}", deviceCode);
                    connection.commit(); // 提交事务
                    break; // 成功退出循环
                } catch (SQLException e) {
                    if (e.getErrorCode() == DEADLOCK_ERROR_CODE && retryCount < 3) {
                        retryCount++;
                        log.warn("死锁重试 {} 次: {}", retryCount, e.getMessage());
                        continue; // 重试事务
                    } else {
                        connection.rollback();
                        log.error("数据库操作失败，已回滚事务。错误信息: {}", e.getMessage(), e);
                        throw e;
                    }
                }
            }

        }
    }

    /**
     *  删除指定 deviceCode 的数据
     * @param connection
     * @param deviceCode
     * @return
     * @throws SQLException
     */
    private boolean deleteIfExists(Connection connection,  String deviceCode) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseSql)) {
            stmt.setString(1, deviceCode);
            int affectedRows = stmt.executeUpdate();
            return affectedRows > 0; // 如果删除了数据，返回 true
        }
    }

    /**
     *  批量插入或更新数据
     * @param connection
     * @param deviceContainerDOs
     * @return
     * @throws SQLException
     */
    private int insertBatchData(Connection connection, List<DeviceContainerDO> deviceContainerDOs) throws SQLException {
        String sql = MessageFormat.format(INSERT_SQL, tableName);
        try (PreparedStatement stmt = connection.prepareStatement(sql)) {
            int batchSize = 0;

            for (DeviceContainerDO item : deviceContainerDOs) {
                try {
                    stmt.setString(1, item.getId());
                    stmt.setString(2, item.getContainerName());
                    stmt.setString(3, item.getImageVersion());
                    stmt.setString(4, item.getCreateStatus());
                    stmt.setString(5, item.getContainerStatus());
                    stmt.setString(6, item.getDeviceCode());
                    stmt.setString(7, item.getIsRemove());
                    stmt.setString(8, item.getCpuRate());
                    stmt.setString(9,item.getMemoryRate());
                    stmt.setString(10,item.getStorageRate());
                    stmt.setLong(11,item.getSource());
                    stmt.setString(12,item.getIp());
                    stmt.setDate(13, new java.sql.Date(item.getStarted().getTime()));
                    stmt.setLong(14,item.getLifeTime());
                    stmt.setString(15,item.getImage());
                    stmt.addBatch();
                    if (++batchSize % 10 == 0) { // 每5条提交一次
                        stmt.executeBatch();
                    }
                }catch (SQLException e) {
                    log.error("插入或更新记录时出错，跳过。ID：{}，错误：{}", item.getId(), e.getMessage(), e);
                }
            }
            stmt.executeBatch(); // 插入剩余数据
            return batchSize;
        }
    }

    @Override
    public void close() throws Exception {
        rdsUtils.close();
        super.close();
    }
}
